package com.simoniu.flink.java.tablesql;

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import java.util.ArrayList;
import static org.apache.flink.table.api.Expressions.$;

/**
 * 将DataSet转换成表
 * Created by simoniu
 */
public class DataSetToTableJavaDemo {
    public static void main(String[] args) {
        //获取BatchTableEnvironment
        ExecutionEnvironment bbEnv = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment bbTableEnv = BatchTableEnvironment.create(bbEnv);

        //获取DataSet
        ArrayList<Tuple3<Integer, String, String>> data = new ArrayList<>();
        data.add(new Tuple3<Integer, String, String>(1, "jack", "male"));
        data.add(new Tuple3<Integer, String, String>(2, "tom", "female"));
        data.add(new Tuple3<Integer, String, String>(3, "mike", "male"));
        DataSource<Tuple3<Integer, String, String>> set = bbEnv.fromCollection(data);

        //第一种：将DataSet转换为view视图
        bbTableEnv.createTemporaryView("myTable", set, $("id"), $("name"), $("gender"));
        bbTableEnv.sqlQuery("select * from myTable where id > 1").execute().print();

        //第二种：将DataSet转换为table对象
        Table table = bbTableEnv.fromDataSet(set, $("id"), $("name"), $("gender"));
        table.select($("id"), $("name"), $("gender"))
                .filter($("id").isGreater(1))
                .execute()
                .print();
    }
}
